package main

import (
	"flag"
	"fmt"
	"net/http"
	"os"
	"os/signal"
	"strconv"
	"syscall"

	"github.com/upmio/horus-polymerizer/api"
	"github.com/upmio/horus-polymerizer/comsumer"
	"github.com/upmio/horus-polymerizer/log"
	"github.com/upmio/horus-polymerizer/queue"
	"github.com/upmio/horus-polymerizer/receiver"
	"github.com/upmio/horus-polymerizer/utils"

	"encoding/json"
	"time"
)

var (
	flIP            = flag.String("ip", "0.0.0.0", " usage: the srv ip")
	flPort          = flag.Int("port", 1234, "usage: the srv port")
	flCouchbaseAddr = flag.String("cb", "192.168.2.131,192.168.2.132", "the address of the couchbase cluster")
	flmysqlSource   = flag.String("mysql", "", "the address of the mysql(eg:root:123456@tcp(127.0.0.1:3306)/Test?charset=utf8)")
	// Node  = flag.String("node", "", "the host_polymerizer belong to which consul node Agent")
	flHsrv  = flag.String("hsrv", "", "the ip:port of the horus_server; eg:192.168.2.1:8111#192.168.2.2:8111")
	flCheck = flag.String("check", "health", "tell hsrv how stragy to check health")

	flEventQueuesLen    = flag.Int("eventlen", 30000, " usage: the event queue len")
	flEventQueuesDisNum = flag.Int("eventdisnum", 2000, "usage : if the event queues has full,to discard the numbers of the datas")
	flEventComsumerNum  = flag.Int("eventcomsumerNum", 2, "usage: the event comsumer numss")

	flMetricQueuesLen    = flag.Int("metriclen", 300000, " usage: the metric queue len")
	flMetricQueuesDisNum = flag.Int("metricdisnum", 2000, "usage : if the metric queues has full,to discard the numbers  of the datas")
	flMetricComsumerNum  = flag.Int("metriccomsumerNum", 6, "usage: the Metric comsumer numss")

	flLoglevel = flag.String("loglevel", "info", " the log level")
	flLogfile  = flag.String("logfile", "/var/log/horus-polymerizer.log", "the log file ")
	flDebug    = flag.Bool("debug", false, "debug model ,write log to std.err")

	flDebugPort = flag.Int("dport", 4321, "usage: the  debug port for pprof")

	flConsulIP         = flag.String("consulIP", "0.0.0.0", " usage: the Consul IP")
	flConsulDatecenter = flag.String("datacenter", "", " usage: the datacenter of the consul ")
	// Metricfile = flag.String("mfile", "/etc/horusplyer", "the file define what metrics don't need history")
	flVersion = flag.Bool("version", false, "show the version ")
	flTTL     = flag.Int("ttl", 50, "the ttl of the plyserver config in the consul")

	flEnableKafka      = flag.Bool("enable-kafka", false, "send metrics datas to kafka")
	flKafkaMetrictopic = flag.String("kafka-metric-topic", "dbass_metric_topic", "kafka metric topic to send")
	flKafkaEventtopic  = flag.String("kafka-event-topic", "dbass_event_topic", "kafka event topic to send")
	flkafkaAddrs       = flag.String("kafka-addr", "145.4.232.211:9092,145.4.232.213:9092", "kafka addr")
)

var (
	version = "v4.0.0"
	build   = "no set"
)

func registerToConsul(ip string, port int) error {

	appname := utils.GetPlymerName(ip, strconv.Itoa(port))
	// scripts := "/usr/local/consul_scripts/polymerizer.sh"
	tags := []string{"plymer"}

	if err := utils.RegisterTCPToCluster(appname, ip, port, 60, 30, tags); err != nil {
		log.Error("register to consul fail:", err)
		return err
	}

	log.Info(appname, ":Register the plyers to consul ok.")

	return nil
}

func putPlyersKey(ip string, port int) error {
	appname := utils.GetPlymerName(ip, strconv.Itoa(port))
	key := utils.PlyerConsulDir + appname

	ply := utils.Plymer{
		Pro:  "tcp",
		IP:   ip,
		Port: strconv.Itoa(port),
		Name: appname,
	}

	plybyte, err := json.Marshal(ply)
	if err != nil {
		return fmt.Errorf(" %s json.Marshal fail:%v", ply, err)
	}

	if err := utils.PutKeyToClusterWithTTL(key, plybyte, time.Duration(*flTTL)); err != nil {
		return fmt.Errorf(" %s PutKeyToClusterWithTTl fail:%v", ply, err)
	}

	go plyersKeyManagerWithTTL(key, plybyte, *flTTL)

	return nil
}

func plyersKeyManagerWithTTL(key string, value []byte, ttl int) {
	log.Info("start  plyersKeyManagerWithTTL goroutine.")

	for {
		time.Sleep(time.Duration(ttl-10) * time.Second)
		log.Debug("reput key with TTL do again")

		if err := utils.PutKeyToClusterWithTTL(key, value, time.Duration(*flTTL)); err != nil {
			log.Errorf("  %s PutKeyToClusterWithTTl fail:%v", key, err)
		}
	}

}

func unregister(serviveID string) error {
	err := utils.UnRegisterFromCluster(serviveID)
	if err != nil {
		return err
	}

	log.Info(serviveID, ":Unregister from consul ok")

	return nil
}

func main() {

	flag.Parse()

	if *flVersion {
		fmt.Println("the version is :", version)
		fmt.Println("the build is :", build)
		return
	}

	logfd, err := os.OpenFile(*flLogfile, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0666)
	if err != nil {
		fmt.Println("open logfile fail")
		os.Exit(1)
	}
	defer logfd.Close()

	if err := log.LogInit(*flLoglevel, logfd); err != nil {
		fmt.Println("log init fail", err)
		os.Exit(1)
	}

	if *flmysqlSource == "" || *flHsrv == "" || *flIP == "0.0.0.0" || *flConsulIP == "0.0.0.0" || *flConsulDatecenter == "" || *flTTL <= 10 {
		log.Error("the mysql,IpFlag,ConsulIp,ConsulDatecenter  and Hsrv args must be set and flTTL must bigger than 10 ")
		os.Exit(1)
	}

	if err := utils.Initconsulcfg(*flConsulIP, *flConsulDatecenter); err != nil {
		log.Error("Initconsulcfg fail :", err)
		os.Exit(2)
	}

	//queue init
	if err := queue.InitEventQueue(*flEventQueuesLen, *flEventQueuesDisNum); err != nil {
		log.Error("init EventQueue fail ,use defuat args.err:", err)
		os.Exit(2)
	}

	if err := queue.InitMetricQueue(*flMetricQueuesLen, *flMetricQueuesDisNum); err != nil {
		log.Error("init MetricQueue fail ,use defuat args.err:", err)
		os.Exit(2)
	}
	port := strconv.Itoa(*flPort)

	//comsumer init
	event := comsumer.NewEventComsumerManager(*flmysqlSource, *flEventComsumerNum)
	event.InitAndStartManager()

	metric := comsumer.NewMetricComsumerManager(*flCouchbaseAddr, "monitor", "", *flMetricComsumerNum)

	if err := metric.InitAndStartManager(); err != nil {
		log.Error("InitAndStart metric Manager fail:", err)
		os.Exit(2)
	}

	if *flEnableKafka {
		kafkaComsumer := comsumer.NewKafkaMetricComsumer(*flkafkaAddrs, *flKafkaEventtopic, *flKafkaMetrictopic)
		go kafkaComsumer.StartManager()
	}

	// server register
	if err := registerToConsul(*flIP, *flPort); err != nil {
		log.Errorf("register to consul fail: %v", err)
		os.Exit(2)
	}

	if err := putPlyersKey(*flIP, *flPort); err != nil {
		log.Errorf("PutPlyersKey to consul fail: %v", err)
		os.Exit(2)
	}

	defer func() {
		appname := utils.GetPlymerName(*flIP, port)
		key := utils.PlyerConsulDir + appname

		log.Infof("try to remove the PlyersKey:%s", key)

		if err := utils.DelKeyFromCluster(key); err != nil {
			log.Errorf("del the PlyersKey:%s fail: %v", key, err)
		} else {
			log.Infof("del the PlyersKey:%s ok", key)
		}
	}()

	if err := utils.ComponentRegisterToServer(
		utils.GetPlymerName(*flIP, port),
		utils.GetPlymerName(*flIP, port),
		"plymer",
		*flCheck, *flHsrv); err != nil {

		log.Errorf("Component Register To horus server :%v", err)
		os.Exit(2)
	}
	log.Info("Component Register To horus server ok.")

	server := receiver.NewHproseTCPServer(*flIP, port)

	//signal accept
	sigs := make(chan os.Signal, 1)
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)

	go func() {
		<-sigs
		log.Info("accept a SIGINT,Unregister the server from consul and hosrus-server")

		if err := utils.ComponentUnRegisterToServer(
			utils.GetPlymerName(*flIP, port), *flHsrv); err != nil {

			log.Errorf("Component UnRegister to horus-server fail :%v", err)
		}

		if err := unregister(utils.GetPlymerName(*flIP, port)); err != nil {

			log.Errorf("unregister fail:%v. and not unregiester to horus-server", err)
		} else {

			if err := utils.ComponentUnRegisterToServer(
				utils.GetPlymerName(*flIP, port), *flHsrv); err != nil {

				log.Errorf("Component UnRegister to horus-server fail :%v", err)
			}
		}

		os.Exit(0)
	}()

	//debug for pprof
	if *flDebug {
		r := api.NewDebugRoute()
		srvURI := "0.0.0.0" + ":" + strconv.Itoa(*flDebugPort)

		go http.ListenAndServe(srvURI, r)

		fmt.Printf("pprof debug listern on: %s", srvURI)
	}

	//server start now
	server.Start()
}
